%% copy riak-0.14.2 source to learn dynamo implementation and erlang
%% author: lichuang
%% copy source from riak_client.erl

%% riak_client: object used for access into the riak system

-module(edynamo_client, [Node, ClientId]).
-export([put/1, put/2, put/3, put/4, put/5]).
-export([get/2, get/3,get/4]).
-export([set_bucket/2, get_bucket/1]).

%% @type default_timeout() = 60000
-define(DEFAULT_TIMEOUT, 60000).
-define(DEFAULT_ERRTOL, 0.00003).
-type edynamo_client() :: term().

%% @spec put(RObj :: edynamo_object:edynamo_object()) ->
%%        ok |
%%       {error, too_many_fails} |
%%       {error, timeout} |
%%       {error, {n_val_violation, N::integer()}}
%% @doc Store RObj in the cluster.
%%      Return as soon as the default W value number of nodes for this bucket
%%      nodes have received the request.
%% @equiv put(RObj, W, W, default_timeout())
put(RObj) -> put(RObj, default, default, ?DEFAULT_TIMEOUT).

%% @spec put(RObj :: edynamo_object:edynamo_object(), W :: integer()) ->
%%        ok |
%%       {error, too_many_fails} |
%%       {error, timeout} |
%%       {error, {n_val_violation, N::integer()}}
%% @doc Store RObj in the cluster.
%%      Return as soon as at least W nodes have received the request.
%% @equiv put(RObj, W, W, default_timeout())
put(RObj, W) -> put(RObj, W, W, ?DEFAULT_TIMEOUT).

%% @spec put(RObj::edynamo_object:edynamo_object(),W :: integer(),DW :: integer()) ->
%%        ok |
%%       {error, too_many_fails} |
%%       {error, timeout} |
%%       {error, {n_val_violation, N::integer()}}
%% @doc Store RObj in the cluster.
%%      Return as soon as at least W nodes have received the request, and
%%      at least DW nodes have stored it in their storage backend.
%% @equiv put(Robj, W, DW, default_timeout())
put(RObj, W, DW) -> put(RObj, W, DW, ?DEFAULT_TIMEOUT).

%% @spec put(RObj::edynamo_object:edynamo_object(), W :: integer(), DW :: integer(),
%%           TimeoutMillisecs :: integer()) ->
%%        ok |
%%       {error, too_many_fails} |
%%       {error, timeout} |
%%       {error, {n_val_violation, N::integer()}}
%% @doc Store RObj in the cluster.
%%      Return as soon as at least W nodes have received the request, and
%%      at least DW nodes have stored it in their storage backend, or
%%      TimeoutMillisecs passes.
put(RObj, W, DW, Timeout) -> put(RObj, W, DW, Timeout, []).

%% @spec put(RObj::edynamo_object:edynamo_object(), W :: integer(), DW :: integer(),
%%           TimeoutMillisecs :: integer(), Options::list()) ->
%%        ok |
%%       {error, too_many_fails} |
%%       {error, timeout} |
%%       {error, {n_val_violation, N::integer()}}
%% @doc Store RObj in the cluster.
%%      Return as soon as at least W nodes have received the request, and
%%      at least DW nodes have stored it in their storage backend, or
%%      TimeoutMillisecs passes.
put(RObj, W, DW, Timeout, Options) ->
  R0 = edynamo_object:increment_vclock(RObj, ClientId),
  Me = self(),
  ReqId = mk_reqid(),
  spawn(Node, edynamo_put_fsm, start, [ReqId, R0, W, DW, Timeout, Me, Options]),
  wait_for_reqid(ReqId, Timeout).

%% @spec get(edynamo_object:bucket(), edynamo_object:key()) ->
%%       {ok, edynamo_object:edynamo_object()} |
%%       {error, notfound} |
%%       {error, timeout} |
%%       {error, {n_val_violation, N::integer()}} |
%%       {error, Err :: term()}
%% @doc Fetch the object at Bucket/Key.  Return a value as soon as the default
%%      R-value for the nodes have responded with a value or error.
%% @equiv get(Bucket, Key, R, default_timeout())
get(Bucket, Key) -> get(Bucket, Key, default, ?DEFAULT_TIMEOUT).

%% @spec get(edynamo_object:bucket(), edynamo_object:key(), R :: integer()) ->
%%       {ok, edynamo_object:edynamo_object()} |
%%       {error, notfound} |
%%       {error, timeout} |
%%       {error, {n_val_violation, N::integer()}} |
%%       {error, Err :: term()}
%% @doc Fetch the object at Bucket/Key.  Return a value as soon as R
%%      nodes have responded with a value or error.
%% @equiv get(Bucket, Key, R, default_timeout())
get(Bucket, Key, R) -> get(Bucket, Key, R, ?DEFAULT_TIMEOUT).

%% @spec get(edynamo_object:bucket(), edynamo_object:key(), R :: integer(),
%%           TimeoutMillisecs :: integer()) ->
%%       {ok, edynamo_object:edynamo_object()} |
%%       {error, notfound} |
%%       {error, timeout} |
%%       {error, {n_val_violation, N::integer()}} |
%%       {error, Err :: term()}
%% @doc Fetch the object at Bucket/Key.  Return a value as soon as R
%%      nodes have responded with a value or error, or TimeoutMillisecs passes.
get(Bucket, Key, R, Timeout) when is_binary(Bucket), is_binary(Key),
                                  (is_atom(R) or is_integer(R)),
                                  is_integer(Timeout) ->
  Me = self(),
  ReqId = mk_reqid(),
  spawn(Node, edynamo_get_fsm, start, [ReqId, Bucket, Key, R, Timeout, Me]),
  wait_for_reqid(ReqId, Timeout).

%% @private
mk_reqid() ->
  erlang:phash2(erlang:now()). % only has to be unique per-pid

%% @private
wait_for_reqid(ReqId, Timeout) ->
  receive
    {ReqId, {error, Err}} -> {error, Err};
    {ReqId, ok} -> ok;
    {ReqId, {ok, Res}} -> {ok, Res}
  after Timeout ->
      {error, timeout}
  end.

%% @spec set_bucket(edynamo_object:bucket(), [BucketProp :: {atom(),term()}]) -> ok
%% @doc Set the given properties for Bucket.
%%      This is generally best if done at application start time,
%%      to ensure expected per-bucket behavior.
%% See riak_core_bucket for expected useful properties.
set_bucket(BucketName,BucketProps) ->
  rpc:call(Node, edynamo_bucket, set_bucket, [BucketName,BucketProps]).

%% @spec get_bucket(edynamo_object:bucket()) -> [BucketProp :: {atom(),term()}]
%% @doc Get all properties for Bucket.
%% See riak_core_bucket for expected useful properties.
get_bucket(BucketName) ->
  rpc:call(Node, edynamo_bucket, get_bucket,[BucketName]).
